package com.ct.ecommerce.base.service;

import com.ct.ecommerce.api.base.client.DelayMessageService;
import com.ct.ecommerce.api.base.dto.DelayMessageDTO;
import com.ct.ecommerce.base.core.DelayLevelCalculate;
import com.ct.ecommerce.base.core.SendRealMessageTask;
import com.ct.ecommerce.base.core.TimeWheelFactory;
import com.ct.ecommerce.framework.constant.DelayMessageConstants;
import com.ct.ecommerce.framework.constant.MqConstants;
import com.google.common.base.Preconditions;
import io.netty.util.HashedWheelTimer;
import java.util.Date;
import java.util.Objects;
import java.util.StringJoiner;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboService;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

/**
 * @author changtong
 * @since 2021/6/21
 */
@DubboService
@Slf4j
public class DelayMessageServiceImpl implements DelayMessageService {

    @Resource
    private RocketMQTemplate mqTemplate;

    @Override
    public Boolean sendDelay(DelayMessageDTO delayMessage, Date startSendTime) throws Exception {
        long time = startSendTime.getTime();
        long now = System.currentTimeMillis();
        long l = (time - now) / 1000;
        log.info("进入.............sendDelay....... time = {} , now = {} , l = {} 秒",time,now,l);

        if (l <= DelayMessageConstants.TIME_OUT) {
            //小于60S，直接放入内存HashedWheelTimer执行发送
            HashedWheelTimer instance = TimeWheelFactory.getInstance();
            CountDownLatch countDownLatch = new CountDownLatch(1);
            SendRealMessageTask sendRealMessageTask
                = new SendRealMessageTask(mqTemplate, delayMessage, countDownLatch);
            instance.newTimeout(sendRealMessageTask, l < 0 ? 1 : l, TimeUnit.SECONDS);
            countDownLatch.await();
            return Objects.nonNull(sendRealMessageTask.getSendResult())
                ? sendRealMessageTask.getSendResult().getSendStatus() == SendStatus.SEND_OK
                : Boolean.FALSE;
        } else {
            Integer level = DelayLevelCalculate.calculateDefault(l);
            Message<DelayMessageDTO> msg = fillMessage(delayMessage, level, startSendTime);
            String destination = new StringJoiner(":").add(MqConstants.MQ_TOPIC_BASE)
                .add(MqConstants.MQ_TAG_BASE_DELAY_CREATE).toString();
            SendResult sendResult = mqTemplate.syncSend(destination, msg, 3000, level);
            return Objects.nonNull(sendResult)
                ? sendResult.getSendStatus() == SendStatus.SEND_OK
                : Boolean.FALSE;
        }
    }

    private Message<DelayMessageDTO> fillMessage(DelayMessageDTO delayMessage, Integer level,
        Date startSendTime) {

        Preconditions.checkNotNull(delayMessage.getTopic(), "message topic is null");
        Preconditions.checkNotNull(delayMessage.getTag(), "message tag is null");
        Preconditions.checkNotNull(delayMessage.getContent(), "message content is null");
        Preconditions.checkNotNull(delayMessage.getKey(), "message content is null");

        return MessageBuilder.withPayload(delayMessage)
            .setHeader(MqConstants.MQ_HEADER_BASE_TIMES,
                String.valueOf(startSendTime.getTime()))
            .build();
    }
}
